//package com.ruoyi.flink.wc;
//
//import com.alibaba.fastjson.JSONObject;
//import com.ruoyi.flink.util.read.ClickHouseRead;
//import com.ruoyi.flink.wc.entity.Student;
//import com.ruoyi.flink.wc.util.RocketMQSource;
//import org.apache.flink.api.common.functions.FlatMapFunction;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.api.java.utils.ParameterTool;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
//import org.apache.flink.table.api.Table;
//import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.apache.flink.util.Collector;
//
///**
// * @Author syc
// * @Date Created in 17:26 2022/5/7
// * @Description
// */
//public class TestFlinkClickHouseRead {
//
//    public static void main(String[] args) throws Exception {
//
//        // System.setProperty("rocketmq.client.logLevel", "ERROR");
//        // 参数中获取值
//        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
//        // 获取 rocketmq的地址端口  "192.168.1.125:9876,192.168.1.125:9875"
//        String namesrvaddr = parameterTool.get("namesrvaddr", "192.168.1.123:9876");
//        // 获取主题(topic)
//        String topic = parameterTool.get("topic", "flink-test");
//        // 获取分组(group)
//        String group = parameterTool.get("flinkGroup", "flink-group");
//        // 创建流式执行环境
//        StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
//        // 设置并行度(默认就是以)
//        env.setParallelism(1);
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        // 创建表环境
//        //  StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        // 获取mq中的参数值
//        DataStreamSource<String> mqSource = env.addSource(new RocketMQSource(namesrvaddr, topic, group));
//        String sql = "select name,age,sex from default.student";
//        DataStreamSource<Student> studentRead = env.addSource(new ClickHouseRead(sql));
//        SingleOutputStreamOperator<Student> wordAndOne =  mqSource.flatMap(new FlatMapFunction<String, Student>() {
//            @Override
//            public void flatMap(String sourc, Collector<Student> out) {
//                JSONObject json = JSONObject.parseObject(sourc);
//                Integer age = json.getInteger("age");
//                Student stu = new Student();
//                stu.age = age;
//                stu.sex = "男";
//                stu.name = "王长江";
//                System.out.println(stu.toString() + "...............");
//                out.collect(stu);
//            }
//        }).connect(studentRead).flatMap(new CoFlatMapFunction<Student, Student, Student>() {
//
//                    @Override
//                    public void flatMap1(Student value, Collector<Student> out) throws Exception {
//                        out.collect(value);
//                    }
//
//                    @Override
//                    public void flatMap2(Student value, Collector<Student> out) throws Exception {
//                        out.collect(value);
//                    }
//                });
//		Table sysmEvent = tEnv.fromDataStream(wordAndOne, "name, age, sex");
//
//        DataStream<Tuple2<Boolean, Student>> mapGridEventDataStream = tEnv.toRetractStream(sysmEvent, Student.class);
//
//        SingleOutputStreamOperator<Student> mapGridEventOutDataStream =
//                mapGridEventDataStream.flatMap(new FlatMapFunction<Tuple2<Boolean,Student>, Student>() {
//
//                    @Override
//                    public void flatMap(Tuple2<Boolean, Student> value, Collector<Student> out) throws Exception {
//                        try {
//                            if(value.f0) {
//
//                                Student student = value.f1;
//
//                                if(student.age.intValue() == 12) {
//                                    out.collect(student);
//                                    System.out.println("student..........." + student.toString());
//                                }
//
//                            }
//                        } catch (Exception ex) {
//
//                        }
//                    }
//                });
//
//
//
//        env.execute();
//    }
//
//
//
//
//}
//
//
//
//
